package com.rstreams;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.streams.core.RocketMQStream;
import org.apache.rocketmq.streams.core.rstream.RStream;
import org.apache.rocketmq.streams.core.rstream.StreamBuilder;
import org.apache.rocketmq.streams.core.rstream.WindowStreamImpl;
import org.apache.rocketmq.streams.core.topology.TopologyBuilder;
import org.apache.rocketmq.streams.core.util.Pair;
import org.apache.rocketmq.streams.core.window.Time;
import org.apache.rocketmq.streams.core.window.WindowInfo;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Properties;

/**
 * @author: LCG
 * @date: 2023-02-15 17:29:44
 * @description:
 **/
public class TestStreamWindows {
    public static void main(String[] args) {

        StreamBuilder streamBuilder=new StreamBuilder("JOb02");
        WindowInfo windowInfo=new WindowInfo();
        windowInfo.setWindowSize(Time.seconds(10));
        windowInfo.setWindowType(WindowInfo.WindowType.TUMBLING_WINDOW);

        streamBuilder.source("TEST-15", msg -> {
            String value = new String(msg, StandardCharsets.UTF_8);
            return new Pair<>(null, value);
        }).map(mm -> {
           MUser mUser = JSON.parseObject(mm, MUser.class);
           mUser.setCreateTime(new Date());
           return mUser;
        }).keyBy(x->{
           return x.getWordName();
        }).window(windowInfo)
                .count()
                .toRStream()
                .print();

        TopologyBuilder topologyBuilder = streamBuilder.build();
        Properties properties = new Properties();
        properties.put(MixAll.NAMESRV_ADDR_PROPERTY, "192.168.137.12:9876");
        RocketMQStream rocketMQStream=new RocketMQStream(topologyBuilder,properties);
        rocketMQStream.start();
    }
}
